package com.shujia.spark.opt

import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo5MapJoin {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("cache")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    import spark.implicits._

    val studentDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING, name STRING ,age INT,gender STRING ,clazz STRING")
      .load("data/students.txt")

    val scoreDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("sid STRING , cid STRING ,sco DOUBLE")
      .load("data/score.txt")

    /**
     * 当大表关联小表（100M）时，可以将小表广播出去，使用map join ， 不会产生shuffle,效率高
     * 广播变量会将数据房子Executor的内存中，如果数据量太大会导致内存溢出
     *
     * 广播变量流程
     * 1、先将小表拉取到Driver端，会产生一个job
     * 2、在将小表广播到Executor端
     * 3、大表可以使用map算子实现join
     */
    studentDF
      .hint("broadcast") //将小表广播出去,在map端进行join
      .join(scoreDF, $"id" === $"sid")
      .show()


    /**
     *
     * sql 实现map join
     * /*+broadcast(b) */ 广播小表
     */

    studentDF.createOrReplaceTempView("student")
    scoreDF.createOrReplaceTempView("score")

    spark.sql(
      """
        |select /*+broadcast(b) */ * from
        |score as a
        |join
        |student as b
        |on a.sid=b.id
        |
        |""".stripMargin)
      .show()


    while (true) {}

  }

}
